package org.trade.common.rocketmq;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.trade.common.exception.AceMQException;

public class AceMQConsumer {
	public static final Logger logger = LoggerFactory.getLogger(AceMQConsumer.class);

	private String namesrvAddr;

	private String topic;

	private String tags = "*";// 多个tag以|分割

	private String groupName;

	// 用户的行为
	private IMessageProcessor processor;

	public void init() throws AceMQException {
		if (StringUtils.isBlank(groupName) || StringUtils.isBlank(topic))
			throw new AceMQException("not be blank");
		// 订阅主题
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(this.groupName);
		consumer.setNamesrvAddr(this.namesrvAddr);
		try {
			//  从头还是从尾开始
			consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
			consumer.subscribe(this.topic, this .tags);
			/**
			 * listener作为一个监听器MessageListenerConcurrently一直监听消费者的是否得到消息，
			 * 一旦得到消息就开始处理
			 */
			AceMessageListener aceMessageListener = new AceMessageListener();
			aceMessageListener.setProcessor(processor);
			consumer.registerMessageListener(aceMessageListener);
			consumer.start();
			logger.info("consumer is start!groupName:{},topic:{},namesrvAddr:{}", topic, topic, namesrvAddr);
		} catch (MQClientException e) {
			logger.error("consumer is start!groupName:{},topic:{},namesrvAddr:{}", topic, topic, namesrvAddr);
			throw new AceMQException(e);
		}

	}

	public void setGroupName(String groupName) {
		this.groupName = groupName;
	}

	public void setNamesrvAddr(String namesrvAddr) {
		this.namesrvAddr = namesrvAddr;
	}

	public void setTopic(String topic) {
		this.topic = topic;
	}

	public void setTags(String tags) {
		this.tags = tags;
	}

	public void setProcessor(IMessageProcessor processor) {
		this.processor = processor;
	}

}
